package com.atguigu.flink09;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/2/29
 * 该案例演示了算子状态----列表状态
 * 需求：在map算子中计算每个并行子任务上数据的个数
 *
 */
public class Flink01_opeState_ListState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> socketDS = env
                .socketTextStream("hadoop102", 8888);

        socketDS.process(
                new MyProF()
        ).print();

        env.execute();
    }
}

class MyProF extends ProcessFunction<String, String> implements CheckpointedFunction {
    //普通变量的作用范围，也是算子的并行子任务
    int count = 0;

    //声明状态
    ListState<Integer> countListState;

    @Override
    public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
        ++count;
        out.collect("当前并行度上元素的个数为" + count);
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        System.out.println("~~~initializeState~~~");
        countListState.clear();
        countListState.add(count);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        System.out.println("~~~initializeState~~~");
        //和键控状态不一样，算子状态的初始化是在initializeState方法中完成的
        ListStateDescriptor<Integer> listStateDescriptor
                = new ListStateDescriptor<Integer>("countListState",Integer.class);
        countListState = context.getOperatorStateStore().getListState(listStateDescriptor);

        if(context.isRestored()){
            //如果刚恢复状态，将状态中的数据赋值给计数变量
            count = countListState.get().iterator().next();
        }

    }
}
